package com.atguigu.gmall.realtime.app.newapp

import com.atguigu.gmall.realtime.app.newapp.DwFavorApp.sparkConf
import com.atguigu.gmall.realtime.bean._
import com.atguigu.gmall.realtime.handler.JoinHandler.JoinType
import com.atguigu.gmall.realtime.handler.{CommonApp, DimHandler, EsHandler, JoinHandler}
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream

import java.time.LocalDate

object DwOrderWideNewApp extends CommonApp {


  //0 环境
  //1 取 偏移量 消费kafka  *2  主表 +从表
  //2 提取偏移量  *2  主表 +从表
  //3 转型成专用的结构化对象 OrderInfo  OrderDetail
  //4 流join  主表从表合并
  //5  补充维度信息  // 用户 地区  商品
  //6  写es
  def main(args: Array[String]): Unit = {
    //0 环境
    val orderInfoTopic = "DWD_ORDER_INFO_I"
    val orderDetailTopic = "DWD_ORDER_DETAIL_I"

    val orderDetailCouponTopic = "DWD_ORDER_DETAIL_COUPON_I"
    val orderDetailActivityTopic = "DWD_ORDER_DETAIL_ACTIVITY_I"
    val groupId = "dw_order_wide_app"

    val ssc: StreamingContext = initSsc(groupId, "local[4]", 5)

    val orderInfoDstream: DStream[OrderInfo] = getDstream[OrderInfo](ssc, orderInfoTopic, groupId)
    val orderDetailDstream: DStream[OrderDetail] = getDstream[OrderDetail](ssc, orderDetailTopic, groupId)
    val orderDetailCouponDstream: DStream[OrderDetailCoupon] = getDstream[OrderDetailCoupon](ssc, orderDetailCouponTopic, groupId)
    val orderDetailActivityDstream: DStream[OrderDetailActivity] = getDstream[OrderDetailActivity](ssc, orderDetailActivityTopic, groupId)
    //   orderDetailCouponDstream.print(1000)
    //
    //   orderInfoDstream.print()
    val orderInfoWithPDStream: DStream[OrderInfo] = DimHandler.joinProvince[OrderInfo](orderInfoDstream, { orderInfo: OrderInfo => orderInfo.province_id.toString })
    val orderInfoWithUDStream: DStream[OrderInfo] = DimHandler.joinUser[OrderInfo](orderInfoWithPDStream, { orderInfo: OrderInfo => orderInfo.user_id.toString })

    //   orderInfoWithUDStream.print()
    val orderInfoWithKDstream: DStream[(String, OrderInfo)] = orderInfoWithUDStream.map(orderInfo => (orderInfo.id.toString, orderInfo))
    val orderDetailWithKDstream: DStream[(String, OrderDetail)] = orderDetailDstream.map(orderDetail => (orderDetail.order_id.toString, orderDetail))
    //
    val orderDetailCouponWithOidDstream: DStream[(String, OrderDetailCoupon)] = orderDetailCouponDstream.map(orderDetailCoupon => (orderDetailCoupon.order_detail_id.toString, orderDetailCoupon))
    val orderDetailActivityWithOidDstream: DStream[(String, OrderDetailActivity)] = orderDetailActivityDstream.map(orderDetailActivity => (orderDetailActivity.order_detail_id.toString, orderDetailActivity))

    //    orderDetailActivityWithOidDstream.print(1000)

    //
    val orderWideNewWithIdDstream: DStream[(String, OrderWideNew)] = JoinHandler.join[String, OrderInfo, OrderDetail, (String, OrderWideNew)](orderInfoWithKDstream, orderDetailWithKDstream
      , JoinType.Inner, 10, "order_join_detail", { case (orderInfoOpt, orderDetailOpt) =>
        val orderWideNew = new OrderWideNew()
        orderWideNew.mergeOrderInfo(orderInfoOpt.get)
        orderWideNew.mergeOrderDetail(orderDetailOpt.get)

        (orderWideNew.detail_id.toString, orderWideNew)
      })

    //   orderWideNewWithIdDstream.print(1000)

    val orderWideNewWithCouponDstream: DStream[(String, OrderWideNew)] = JoinHandler.join[String, OrderWideNew, OrderDetailCoupon, (String, OrderWideNew)](orderWideNewWithIdDstream, orderDetailCouponWithOidDstream
      , JoinType.Left, 10, "order_join_coupon", { case (orderWideNewOpt: Option[OrderWideNew], orderDetailCouponOpt: Option[OrderDetailCoupon]) =>
        val orderWideNew: OrderWideNew = orderWideNewOpt.get
        if (orderDetailCouponOpt != None) {
          orderWideNew.mergeOrderCoupon(orderDetailCouponOpt.get)
        }
        (orderWideNew.detail_id.toString, orderWideNew)
      })

    //

    val orderWideNewWithActivityDstream: DStream[OrderWideNew] = JoinHandler.join[String, OrderWideNew, OrderDetailActivity, OrderWideNew](orderWideNewWithCouponDstream, orderDetailActivityWithOidDstream
      , JoinType.Left, 10, "order_join_activity", { case (orderWideNewOpt, orderDetailActivityOpt) =>
        val orderWideNew = orderWideNewOpt.get
        if (orderDetailActivityOpt != None) {
          orderWideNew.mergeOrderActivity(orderDetailActivityOpt.get)
        }
        orderWideNew
      })

    //
    val orderWideNewWithSkuDStream: DStream[OrderWideNew] = DimHandler.joinSku[OrderWideNew](orderWideNewWithActivityDstream, { orderWide: OrderWideNew => orderWide.sku_id.toString })

    val orderWideNewWithActDStream: DStream[OrderWideNew] = DimHandler.joinActivity[OrderWideNew](orderWideNewWithSkuDStream, { orderWide: OrderWideNew => orderWide.activity_rule_id.toString })
    // orderWideNewWithActDStream.print(10000)
    val orderWideNewWithCouponDStream: DStream[OrderWideNew] = DimHandler.joinCoupon[OrderWideNew](orderWideNewWithActDStream, { orderWide: OrderWideNew => orderWide.coupon_id.toString })


    //3 转型成专用的结构化对象 OrderInfo  OrderDetail

    orderWideNewWithCouponDStream.cache()
    orderWideNewWithCouponDStream.print(1000)

    EsHandler.saveDStreamToEs[OrderWideNew](orderWideNewWithCouponDStream,
      "gmall2022_order_wide_",
      "id",
      {Unit=>this.commitAllOffsetKafka()}
    )



    ssc.start()
    ssc.awaitTermination()
  }

}
